/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.lvyh.lightframe.sharding.client.strategy;

import com.lvyh.lightframe.sharding.client.common.DataTableInfo;

public class HashShardingStrategy implements ShardingStrategy {

    @Override
    public DataTableInfo doShard(String logicDsKey, String logicTableName, Object shardingValue, int tableShardNum, int dbShardNum) {
        DataTableInfo dataTableInfo = new DataTableInfo();

        if (dbShardNum <= 1 && tableShardNum <= 1) {

            //1、Single database and single table
            return new DataTableInfo(logicDsKey, logicTableName);

        } else if (dbShardNum < 2 && tableShardNum > 1) {

            //2、Single database and shard table
            dataTableInfo.setPhysicalDataSourceKey(logicDsKey);
            if (this.isEmpty(shardingValue)) {
                dataTableInfo.setPhysicalTableName(logicTableName);
            } else {
                int index = computePosIndex(shardingValue.toString(), tableShardNum);
                dataTableInfo.setPhysicalTableName(logicTableName + index);
            }

        } else if (dbShardNum > 1 && tableShardNum > 1) {

            //3、Shard database and shard table
            /**
             * Routing strategy for tables and databases:
             * 1. Intermediate variable = primary key ID% (number of db * number of tables per db)
             * 2. DB = rounding (intermediate variable / number of tables per db)
             * 3. Table = intermediate variable% number of tables per db
             */
            long shardValue = Long.valueOf(shardingValue.toString());
            long midVal = shardValue % (tableShardNum * dbShardNum);
            long dbPosition = midVal / tableShardNum;
            long tablePosition = midVal % tableShardNum;

            String physicalTableName = new StringBuilder().append(logicTableName).append(tablePosition).toString();
            String physicalDataSourceKey = new StringBuilder().append(logicDsKey).append(dbPosition).toString();

            dataTableInfo.setPhysicalTableName(physicalTableName);
            dataTableInfo.setPhysicalDataSourceKey(physicalDataSourceKey);

        } else {

            //4、Shard database and single table
            dataTableInfo.setPhysicalTableName(logicTableName);
            if (this.isEmpty(shardingValue)) {
                dataTableInfo.setPhysicalDataSourceKey(logicDsKey);
            } else {
                int index = computePosIndex(shardingValue.toString(), dbShardNum);
                dataTableInfo.setPhysicalDataSourceKey(logicDsKey + index);
            }

        }
        return dataTableInfo;
    }

    private int computePosIndex(String shardingValue, int shardNum) {
        int h;
        int hash = (h = shardingValue.hashCode()) ^ (h >>> 16);

        int index;
        if (is2Power(shardNum)) {
            index = (shardNum - 1) & hash;
        } else {
            index = Math.floorMod(hash, shardNum);
        }
        return index;
    }

    private boolean isEmpty(Object value) {
        if (value == null || value.toString().equals("0")) {
            return true;
        }

        return false;
    }

    public boolean is2Power(int length) {
        return (length & (length - 1)) == 0;
    }
}
